package com.rrd.dw.mr.orc;

import java.io.IOException;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.ToolRunner;
import org.apache.orc.OrcConf;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcInputFormat;
import org.apache.orc.mapreduce.OrcOutputFormat;

import com.rrd.dw.mr.BaseJob;
import com.rrd.dw.mr.key.ETLpairKey;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;

/**
 * @author liubaoxin 2017-08-28 对各个业务系统用户进行统一整合，具体整合方案见wiki
 */
public class UserBaseMergeMR extends BaseJob {

    public static class UserMergeMap
            extends Mapper<NullWritable, OrcStruct, ETLpairKey, MapWritable> {
        MapWritable mp = new MapWritable();
        private Text tableText = new Text();
        private String tableNmae = "";
        private ETLpairKey ekey = new ETLpairKey();;
        private String firstKey = "";
        private long secondKey = 0l;
        private Date date = null;

        @Override
        protected void setup(
                Mapper<NullWritable, OrcStruct, ETLpairKey, MapWritable>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            FileSplit input = (FileSplit) context.getInputSplit();
            tableNmae = input.getPath().getParent().getParent().getParent()
                .getName();
            if (input.getPath().toString()
                .indexOf("ptp_ym_user_base_attribute_all") > 0) {
                tableNmae = "ptpym_user_base_attribute_all";
            }
            System.out.println("tableNmae=" + tableNmae);
            tableText.set(tableNmae);
        }

        @Override
        protected void map(NullWritable key, OrcStruct value, Context context)
                throws IOException, InterruptedException {
            if (value == null || value.getNumFields() <= 0) {
                return;
            }

            mp.clear();
            mp.put(tableText, tableText);
            if (tableNmae.equals("ptp_pdl_user_base_attribute_all")) {
                ConstructUserMap.pdlConstruct2(mp, value);
            } else if (tableNmae.equals("ptp_rrd_user_base_attribute_all")) {
                ConstructUserMap.rrdConstruct2(mp, value);
            } else if (tableNmae.equals("ptp_ym_user_base_attribute_all")) {
                ConstructUserMap.yimeiConstruct2(mp, value);
            }

            if (mp.get(new Text("pid")) == null) {
                return;
            }
            firstKey = mp.get(new Text("pid")).toString();
            if (firstKey.trim().equals("")) {
                return;
            }
            if (mp.get(new Text("regist_time")) != null) {
                date = DateUtil.formateToDate(
                    ((Text) mp.get(new Text("regist_time"))).toString());
            }
            if (date == null) {
                secondKey = 0;
            } else {
                secondKey = date.getTime();
            }
            secondKey = 200000;
//			System.out.println(secondKey);
            ekey.setKey(firstKey, secondKey);
            context.write(ekey, mp);
        }

    }

    public static class UserMergeReduce
            extends Reducer<ETLpairKey, MapWritable, NullWritable, OrcStruct> {
        private final NullWritable nw = NullWritable.get();
        private MapWritable pdlMp = new MapWritable();
        private MapWritable rrdMp = new MapWritable();
        private MapWritable pxdMp = new MapWritable();
        private MapWritable ymMp = new MapWritable();
        private MapWritable yxMp = new MapWritable();
        private String card_type = "identity";
        private OrcStruct orc = (OrcStruct) OrcStruct
            .createValue(TypeDescription.fromString(getSchema()));
        private MapWritable[] arrMap = null;

        @Override
        protected void reduce(ETLpairKey key, Iterable<MapWritable> vals,
                Reducer<ETLpairKey, MapWritable, NullWritable, OrcStruct>.Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            pdlMp.clear();
            rrdMp.clear();
            pxdMp.clear();
            ymMp.clear();
            yxMp.clear();
            arrMap = new MapWritable[5];
            int i = 0;
            int j = 0;
            for (MapWritable m : vals) {
                j++;
                if (m.size() < 3) {
                    context.getCounter("counter", "szie<3").increment(1);
                    continue;
                }
                if (m.get(
                    new Text("ptp_pdl_user_base_attribute_all")) != null) {
                    if (pdlMp.size() > 1) {
                        continue;
                    }
                    pdlMp.putAll(m);
                    arrMap[i] = pdlMp;
                } else if (m.get(
                    new Text("ptp_rrd_user_base_attribute_all")) != null) {
                    if (rrdMp.size() > 1) {
                        continue;
                    }
                    rrdMp.putAll(m);
                    arrMap[i] = rrdMp;
                }
                if (m.get(new Text("ptp_ym_user_base_attribute_all")) != null) {
                    if (ymMp.size() > 1) {
                        continue;
                    }
                    ymMp.putAll(m);
                    arrMap[i] = ymMp;
                }

                i++;

            }
            if (j > 1) {
                context.getCounter("counter", "merge_app_users").increment(1);
            }
            setNewMap();
            String type = generateUserType();
            String pid = arrMap[0].get(new Text("pid")).toString();
            String user_code = MD5Util.getMd5(pid + card_type);
            String age = etlAge(arrMap[0].get(new Text("age")) == null ? ""
                : arrMap[0].get(new Text("age")).toString());

            orc.setFieldValue(0, new Text(user_code));
            orc.setFieldValue(1, new Text(type));
            orc.setFieldValue(2, new Text(pid));
            orc.setFieldValue(3, (Text) arrMap[0].get(new Text("pid_md5")));
            orc.setFieldValue(4, (Text) arrMap[0].get(new Text("pid_off")));

            orc.setFieldValue(5, new Text(card_type));
            orc.setFieldValue(6, (Text) arrMap[0].get(new Text("mobile")));
            orc.setFieldValue(7, (Text) arrMap[0].get(new Text("mobile_md5")));
            orc.setFieldValue(8, (Text) arrMap[0].get(new Text("mobile_off")));

            orc.setFieldValue(9, (Text) arrMap[0].get(new Text("gender")));
            orc.setFieldValue(10, age.equals("") ? null : new Text(age));
            orc.setFieldValue(11, (Text) arrMap[0].get(new Text("edu")));

            context.write(nw, orc);

        }

        private void setNewMap() {
            for (int i = 1; i < arrMap.length; i++) {
                if (arrMap[i] == null) {
                    continue;
                }
                changeNew(arrMap[0], arrMap[i], new Text("pid"));
                changeNew(arrMap[0], arrMap[i], new Text("mobile"));
                changeNew(arrMap[0], arrMap[i], new Text("gender"));
                changeNew(arrMap[0], arrMap[i], new Text("edu"));
            }
        }

        private void changeNew(MapWritable nw, MapWritable old, Text key) {
            if (nw.get(key) == null) {
                if (old.get(key) != null) {
                    nw.put(key, old.get(key));
                }

            }
        }

        private String etlAge(String age) {
            if (age == null || age.equals("")) {
                return age;
            }
            int n = 0;
            try {
                n = Math.round(Float.parseFloat(age));
                if (n > 80 || n <= 0) {
                    age = "";
                } else {
                    age = String.valueOf(n);
                }
            } catch (Exception e) {
                age = "";
            }
            return age;
        }

        private String generateUserType() {
            String[] arr = new String[] { "0", "0", "0", "0", "0" };
            if (this.pdlMp.size() > 0) {
                arr[4] = "1";
            }
            if (this.rrdMp.size() > 0) {
                arr[3] = "1";
            }

            if (this.ymMp.size() > 0) {
                arr[2] = "1";
            }
            if (this.pxdMp.size() > 0) {
                arr[1] = "1";
            }
            if (this.yxMp.size() > 0) {
                arr[0] = "1";
            }
            return String.format("%s%s%s%s%s", arr);
        }
    }

    public static String getSchema() {

        StringBuilder sb = new StringBuilder();
        sb.append("struct<");
        sb.append(
            "usercode:string,user_type:string,pid:string,pid_md5:string,pid_off:string"
                + ",card_type:string,mobile:string,mobile_md5:string,mobile_off:string,gender:string,"
                + "age:string,edu:string");
        sb.append(">");

        return sb.toString();
    }

    @Override
    public int run(String[] arg0) throws Exception {

        Configuration conf = getConf();
        OrcConf.MAPRED_OUTPUT_SCHEMA.setString(conf, getSchema());
        Job job = Job.getInstance(conf);
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        String dt = conf.get("etl.param.dt", "2017-07-27");
        String pdl_input = "/data/gdm/ptp_pdl_user_base_attribute_all/dt="+dt;
        String rrd_input = "/data/gdm/ptp_rrd_user_base_attribute_all/dt="+dt;
        String ym_input = "/data/gdm/ptp_ym_user_base_attribute_all/dt=2017-09-28";
        String output = conf.get("etl.param.output", "/tmp/fdm/user_base_all2");
        if (dt.equals("")) {
            log.info("请输入日期：etl.param.dt");
            return -1;
        }
        if (output.equals("")) {
            log.info("请输入输出路径：etl.param.output");
            return -1;
        }

        log.info("pdl输入路径:" + pdl_input);
        log.info("rrd输入路径:" + rrd_input);
        log.info("医美输入路径:" + ym_input);
        log.info("输出路径:" + output);

        HadoopUtils.delete(conf, new Path(output));
        job.setInputFormatClass(OrcInputFormat.class);
        job.setOutputFormatClass(OrcOutputFormat.class);
        job.setMapOutputKeyClass(ETLpairKey.class);
        job.setMapOutputValueClass(MapWritable.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(OrcStruct.class);
        OrcInputFormat.setInputPaths(job,
            rrd_input + "," + pdl_input + "," + ym_input);
//		OrcInputFormat.setInputPaths(job, pdl_input );
        OrcOutputFormat.setOutputPath(job, new Path(output));
        job.setJarByClass(this.getClass());
        job.setMapperClass(UserMergeMap.class);
        job.setReducerClass(UserMergeReduce.class);
        job.setPartitionerClass(ETLpairKey.FirstPartitioner.class);
        job.setGroupingComparatorClass(ETLpairKey.MyGroupingComparator.class);

        int stat = job.waitForCompletion(true) ? 0 : 1;
        Counters counters = job.getCounters();
        Counter counter = null;
        counter = counters.findCounter("counter", "mapszie<10");
        System.out.println("map字段数小于10的记录数：" + counter.getValue());
        counter = counters.findCounter("counter", "merge_app_users");
        System.out.println("重复手机号记录数：" + counter.getValue());
        return stat;

    }

    public static void main(String[] args)
            throws IOException, ClassNotFoundException, InterruptedException {
        int n = 0;
        try {
            n = ToolRunner.run(new UserBaseMergeMR(), args);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            log.error(e.getMessage(), e);
        }
        System.exit(n);

    }

}
